package com.behavioranalysis.flinkprogram.dao.output;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.api.java.tuple.Tuple8;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;

import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;

public class JDBCExample {
    public static void main(String[] args) throws Exception {

        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Tuple8<Integer, String, String, String, String, String, String, String>> source
                = environment.createInput(JDBCInputFormat.buildJDBCInputFormat()
                        .setDrivername("com.mysql.jdbc.Driver")
                        .setDBUrl("jdbc:mysql://localhost:3306/spark_project?characterEncoding=utf8&useSSL=true")
                        .setUsername("root")
                        .setPassword("Yangsha130333")
                        .setQuery("select * from task")
                        .finish(),
                new TupleTypeInfo(Tuple8.class, INT_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO,
                        STRING_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO)
        );
        DataSet<Tuple7<String, String, String, String, String, String, String>> sourceMap = source.map(
                new MapFunction<Tuple8<Integer, String, String, String, String, String, String, String>, Tuple7<String, String, String, String, String, String, String>>() {
                    @Override
                    public Tuple7<String, String, String, String, String, String, String> map(Tuple8<Integer, String, String, String, String, String, String, String> value) throws Exception {
                        return new Tuple7<>(String.valueOf(value.f1), "test", "test", "test", "test", "test", String.valueOf(value.f7));
                    }
                }
        );
        sourceMap.print();

        sourceMap.output(JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://localhost:3306/spark_project?characterEncoding=utf8&useSSL=true")
                .setUsername("root")
                .setPassword("Yangsha130333")
                .setQuery("insert into task (task_name,create_time,start_time,finish_time,task_type,task_status,task_param) values (?,?,?,?,?,?,?)")
                .finish());
        environment.execute("JDBCExample");
    }
}
